package com.yixiu.bi.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;


/**
 * 抽象类： 消费者监听程序
 * 外部只需要new 一个 KafkaConsumerListener 对象，重写onMessage方法，在该方法中实现具体的处理逻辑
 */
public abstract  class KafkaConsumerListener {
    /**
     * 抽象方法，处理消息的具体方法
     * @param consumerRecord
     */
    public abstract  void onMessage(ConsumerRecord consumerRecord);

    /**
     * 创建一个处理消息对象
     * @param consumerRecord
     * @return
     */
    public ConsumerThreadHandler newConsumerThreadHandler(ConsumerRecord consumerRecord) {
        return  new ConsumerThreadHandler(consumerRecord);
    }

    /**
     * 处理消息对象
     * 实现了Runnable接口，可以放入线程池中处理消息
     */
    public  class ConsumerThreadHandler implements Runnable {

        // 一条消息记录
        private ConsumerRecord consumerRecord;

        public ConsumerThreadHandler(ConsumerRecord consumerRecord){
            this.consumerRecord = consumerRecord;
        }

        /**
         * 多线程的run方法，实际调用的是 KafkaConsumerListener 的onMessage方法
         */
        public void run() {
            onMessage(consumerRecord);
        }
    }

}



